package mq

import (
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/streadway/amqp"
)

// 定义RabbitMQ对象
type rabbitMQ struct {
	connection   *amqp.Connection
	channel      *amqp.Channel
	queueName    string // 队列名称
	routingKey   string // key名称
	exchangeName string // 交换机名称
	exchangeType string // 交换机类型
	user         string
	passWord     string
	host         string
	port         int
	path         string
	ReceiveChan  <-chan amqp.Delivery
	mu           sync.RWMutex
}

type ConfigRb struct {
	QueueName    string // 队列名称
	RoutingKey   string // key名称
	ExchangeName string // 交换机名称
	ExchangeType string // 交换机类型
	User         string
	PassWord     string
	Host         string
	Port         int
	Path         string
}

// 创建一个新的操作对象
func New(conf ConfigRb) (*rabbitMQ, error) {
	rb := &rabbitMQ{
		queueName:    conf.QueueName,
		routingKey:   conf.RoutingKey,
		exchangeName: conf.ExchangeName,
		exchangeType: conf.ExchangeType,
		user:         conf.User,
		passWord:     conf.PassWord,
		host:         conf.Host,
		port:         conf.Port,
		path:         conf.Path,
		mu:           sync.RWMutex{},
	}
	err := rb.connect()
	return rb, err
}

// 关闭RabbitMQ连接
func (r *rabbitMQ) Close() {
	// 先关闭管道,再关闭链接
	err := r.channel.Close()
	if err != nil {
		fmt.Printf("MQ管道关闭失败:%s \n", err)
	}
	err = r.connection.Close()
	if err != nil {
		fmt.Printf("MQ链接关闭失败:%s \n", err)
	}
}

func (r *rabbitMQ) heartbeat() {
	var err error
	for {
		if r.connection.IsClosed() {
			fmt.Println("链接断开")
			if err = r.connect(); err != nil {
				fmt.Println("重连失败")
				time.Sleep(time.Second)
				continue
			}
			channel, err := r.channel.Consume(r.queueName, "", false, false, false, false, nil)
			if err != nil {
				fmt.Println(err)
				time.Sleep(time.Second)
				continue
			}
			r.ReceiveChan = channel
			fmt.Println("重连成功")
		} else {
			fmt.Println("正常")
		}
		time.Sleep(time.Second)
	}
}

// 链接rabbitMQ
func (r *rabbitMQ) connect() error {
	var err error
	RabbitUrl := fmt.Sprintf("amqp://%s:%s@%s:%d/", r.user, r.passWord, r.host, r.port)
	mqConn, err := amqp.Dial(RabbitUrl)
	if err != nil {
		return errors.New(fmt.Sprintf("MQ打开链接失败:%s \n", err))
	}
	r.connection = mqConn // 赋值给RabbitMQ对象
	mqChan, err := mqConn.Channel()
	if err != nil {
		return errors.New(fmt.Sprintf("MQ打开管道失败:%s \n", err))
	}
	r.channel = mqChan // 赋值给RabbitMQ对象
	return nil
}

// 发送任务
func (r *rabbitMQ) Producer(content <-chan string) error {
	// 用于检查队列是否存在,已经存在不需要重复声明
	_, err := r.channel.QueueDeclarePassive(r.queueName, true, false, false, true, nil)
	if err != nil {
		// 队列不存在,声明队列
		// name:队列名称;durable:是否持久化,队列存盘,true服务重启后信息不会丢失,影响性能;autoDelete:是否自动删除;noWait:是否非阻塞,
		// true为是,不等待RMQ返回信息;args:参数,传nil即可;exclusive:是否设置排他
		_, err = r.channel.QueueDeclare(r.queueName, true, false, false, true, nil)
		if err != nil {
			return errors.New(fmt.Sprintf("MQ注册队列失败:%s \n", err))
		}
	}
	// 队列绑定
	err = r.channel.QueueBind(r.queueName, r.routingKey, r.exchangeName, true, nil)
	if err != nil {
		return errors.New(fmt.Sprintf("绑定队列失败:%s \n", err))
	}
	// 用于检查交换机是否存在,已经存在不需要重复声明
	err = r.channel.ExchangeDeclarePassive(r.exchangeName, r.exchangeType, true, false, false, true, nil)
	if err != nil {
		// 注册交换机
		// name:交换机名称,kind:交换机类型,durable:是否持久化,队列存盘,true服务重启后信息不会丢失,影响性能;autoDelete:是否自动删除;
		// noWait:是否非阻塞, true为是,不等待RMQ返回信息;args:参数,传nil即可; internal:是否为内部
		err = r.channel.ExchangeDeclare(r.exchangeName, r.exchangeType, true, false, false, true, nil)
		if err != nil {
			return errors.New(fmt.Sprintf("MQ注册交换机失败:%s \n", err))
		}
	}
	go r.heartbeat()
	for msg := range content {
		// 发送任务消息
		err = r.channel.Publish(r.exchangeName, r.routingKey, false, false, amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(msg),
		})
		if err != nil {
		}
	}
	return nil
}

// 监听接收者接收任务
func (r *rabbitMQ) Receiver() error {
	// 用于检查队列是否存在,已经存在不需要重复声明
	_, err := r.channel.QueueDeclarePassive(r.queueName, true, false, false, true, nil)
	if err != nil {
		// 队列不存在,声明队列
		// name:队列名称;durable:是否持久化,队列存盘,true服务重启后信息不会丢失,影响性能;autoDelete:是否自动删除;noWait:是否非阻塞,
		// true为是,不等待RMQ返回信息;args:参数,传nil即可;exclusive:是否设置排他
		_, err = r.channel.QueueDeclare(r.queueName, true, false, false, true, nil)
		if err != nil {
			return errors.New(fmt.Sprintf("MQ注册队列失败:%s \n", err))
		}
	}
	// 绑定任务
	err = r.channel.QueueBind(r.queueName, r.routingKey, r.exchangeName, true, nil)
	if err != nil {
		return errors.New(fmt.Sprintf("绑定队列失败:%s \n", err))
	}
	// 获取消费通道,确保rabbitMQ一个一个发送消息
	if err = r.channel.Qos(1, 0, true); err != nil {
		return err
	}
	channel, err := r.channel.Consume(r.queueName, "", false, false, false, false, nil)
	r.ReceiveChan = channel
	if err != nil {
		return errors.New(fmt.Sprintf("获取消费通道异常:%s \n", err))
	}
	go r.heartbeat()
	return nil
}
